package com.beesphere.products.throttling.executor.impl;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beesphere.products.throttling.EnhancedThread;
import com.beesphere.products.throttling.executor.ExecutorServiceFactory;
import com.beesphere.products.throttling.executor.PoolConfiguration;
import com.beesphere.products.throttling.processor.async.ThreadProcessor;

public class DefaultExecutorServiceFactory implements ExecutorServiceFactory {

	private static final long serialVersionUID = -3303979990757376732L;
	
	private static final Logger logger = LoggerFactory.getLogger (ThreadProcessor.class);
	
	public static final String DEFAULT_GROUP = "com.beesphere.throttler.group";
	
	private long maxThreadStackSize;

	@Override
	public ExecutorService create (final PoolConfiguration config) {
		String queueType = config.getQueueType ();
		if (queueType == null) {
			queueType = PoolConfiguration.QUEUE_TYPE_DEQUE;
		}
		BlockingQueue<Runnable> queue = null;
		if (queueType.equals (PoolConfiguration.QUEUE_TYPE_DEQUE)) {
			queue = new LinkedBlockingDeque<Runnable> (config.getTaskQueue ());
		} else if (queueType.equals (PoolConfiguration.QUEUE_TYPE_QUEUE)) {
			queue = new LinkedBlockingQueue<Runnable> (config.getTaskQueue ());
		} else {
			queue = new LinkedBlockingDeque<Runnable> (config.getTaskQueue ());
		}
		
		ThreadPoolExecutor executor = 
        	new ThreadPoolExecutor (
    			config.getCoreSize (), config.getMaxSize (), 
    			config.getKeepAliveTime(), TimeUnit.MILLISECONDS, 
    			queue, 
    			new ThreadFactory () {
    				private AtomicInteger threadId = new AtomicInteger (0);
		            public Thread newThread (Runnable runnable) {
		            	String groupName = config.getGroup ();
		            	if (groupName == null) {
		            		groupName = DEFAULT_GROUP;
		            	}
		            	String threadName = config.getName();
		            	if (threadName == null) {
		            		threadName = config.getGroup ();
		            	}
		            	threadName += "." + System.currentTimeMillis () + "." + threadId.incrementAndGet ();
		            	
		            	long tss = config.getStackSize ();
		            	logger.debug("======Thread " + threadName + " allocated Stack Size: " + tss);
		            	if (tss <= 0 || tss >= maxThreadStackSize) {
		            		tss = PoolConfiguration.THREAD_STACK_SIZE;
		            	}
		            	EnhancedThread thread = new EnhancedThread (
		                    new ThreadGroup (groupName), runnable, threadName, tss);
		                thread.setDaemon (config.isDaemon ());
		                thread.setPriority (config.getPriority ());
		                return thread;
		            }
    			}
        	);
		executor.setRejectedExecutionHandler (new WaitPolicy (config.getWaitTimeout (), TimeUnit.MILLISECONDS));
        return executor;
	}

	public long getMaxThreadStackSize () {
		return maxThreadStackSize;
	}

	public void setMaxThreadStackSize (long maxThreadStackSize) {
		this.maxThreadStackSize = maxThreadStackSize;
	}

}
